package com.foreveross.fetl.queue;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import javax.annotation.Resource;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.log4j.Logger;
import org.springframework.core.task.TaskExecutor;

import com.foreveross.fetl.common.HBaseTable;
import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.core.domain.task.SourceEntity;
import com.foreveross.fetl.queue.util.ExchangeRateUtil;
import com.foreveross.fetl.queue.util.LogMessageUtils;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.RowkeyUtils;
import com.foreveross.fetl.util.StringUtils;

/*-
 * 
 * 触发时间： 每小时一次
 * 
 * 
 *抓取基础数据表(CRAWL_PLANE_FREIGHT)：  
 *rowkey设计
 *抓取基础数据表(CRAWL_PLANE_FREIGHT)：  
 *	 抓取日期(年月日时)_渠道_区域ID_起点(使用三字码)_终点(使用三字码)_航班日期(年月日时)_航班ID_5位随即数
 *	         10位 +    5位  + 2位 +    3位   +         3位      +           10位+            8位 +    5位
 *   
 * 每天的航班历史最低价(即每天抓取多次航班中的最低价格)LP_FLIGHT_HISTORY
 * rowkey: 渠道_区域ID_起点(使用三字码)_终点(使用三字码)_抓取时间(年月日)_航班日期(年月日时)_航班号
 * 
 * 每天的航线历史最低价(即每天抓取多次中的航线下的所有航班的最低价最低价格)
 * 
 */

// @Transactional
public class HistoryDataPreprocess {

	private static Logger log = Logger.getLogger("history");
	
	
	@Resource(name="taskExecutor")
	private  TaskExecutor taskExecutor;
	
	private int tasksNumber;//任务数 多少条rowkey
	
	private int failuresNumber;//失败数
	
	private int threadsNumber;//线程数目
	
	private int executeNumber;//已执行的任务条数
	
	private int processNumber; //process 执行次数
	
	private int insertNumber; //插入条数 
	
	private long excuteStartTime;
	
	
	private String taskTime=null;
	
	
	public void  rollBackNumber(){
		tasksNumber=0;
		failuresNumber=0;
		threadsNumber=0;
		executeNumber=0;
		processNumber=0;
		insertNumber=0;
		excuteStartTime = System.currentTimeMillis();
		LogMessageUtils.create("统计历史最低价", this.getClass().getSimpleName());
	}

	/** 需要统计的列明 */

	public static void main(String[] args) throws Exception {
		HistoryDataPreprocess pro = new HistoryDataPreprocess();
		long startTime = System.currentTimeMillis();
		final Calendar end = Calendar.getInstance();
		final Calendar start = Calendar.getInstance();
		start.set(2013, 4, 2, 9, 0);
		  end.set(2013, 4, 3, 10, 0);
		pro.process(start.getTime(), end.getTime());
		long entTime = System.currentTimeMillis();
		System.out.println(" HistoryDataPreprocessJob 统计共用时(ms)："
				+ (entTime - startTime));

	}

	/**
	 * @throws InterruptedException
	 */
	public void execute() throws InterruptedException {
		Date endTime= new Date();
		rollBackNumber();
		long startTime = System.currentTimeMillis();
		final Calendar c = Calendar.getInstance();
		c.add(Calendar.HOUR_OF_DAY, -1);
		taskTime=DateUtils.format(c.getTime(), "yyyy-MM-dd HH");
		try {
			process(c.getTime(), endTime);
		} catch (Exception e) {
			LogMessageUtils.build("统计历史最低价", this.getClass().getSimpleName(), 3, StringUtils.getStackTrace(e), failuresNumber, tasksNumber, 
					insertNumber, taskTime, excuteStartTime);
		}
		
		long entTime = System.currentTimeMillis();
		log.info(" HistoryDataPreprocessJob 统计共用时(ms)："
				+ (entTime - startTime));
	}
	
	public void manualExecute(final String start,String time){
		this.rollBackNumber();
		try {
		List<String> timeList = Arrays.asList(time.split(";"));
		Calendar c = Calendar.getInstance();
		taskTime = start + " " +time;
		for (String s : timeList) {
			String str =start;
			str += s;
			Date startTime = DateUtils.parse(str, "yyyyMMddHH");
			c.setTime(startTime);
			c.add(Calendar.HOUR_OF_DAY, +1);
			Date endTime = c.getTime();
			this.process(startTime, endTime);
		}
		} catch (Exception e) {
			LogMessageUtils.build("统计历史最低价", this.getClass().getSimpleName(), 3, StringUtils.getStackTrace(e), failuresNumber, tasksNumber, 
					insertNumber, taskTime, excuteStartTime);
			log.error(StringUtils.getStackTrace(e));
		}
	}

	/**
	 * @param startTime
	 * @param endTime
	 * @throws InterruptedException
	 */
	public void process(final Date startTime, final Date endTime) {
		processNumber++;
		if(processNumber ==1){
			ExchangeRateUtil.initExchangeRate(DateUtils.format(new Date(), "yyyyMMdd"));
		}
		SourceEntity source = new SourceEntity();
		List<SourceEntity> listHome=source.findSourceEntityByType(SourceEntity.TYPE_HOME);
		List<SourceEntity> listchannels = source.findSourceEntityByType(SourceEntity.TYPE_CHNNAEL);
		List<String> strchannels= new ArrayList<String>();
		if(null!=listHome){
			for(SourceEntity entity:listHome){
				strchannels.add(entity.getHbaseKey());
			}
		}
		if(null!=listchannels){
			for(SourceEntity entity:listchannels){
				strchannels.add(entity.getHbaseKey());
			}
		}
		Collection<String> channels = strchannels;
		long start = System.currentTimeMillis();
		log.info("HistoryDataPreprocess start time：" + DateUtils.format(startTime, "yyyyMMddHH")
				+ "end time：" + DateUtils.format(endTime, "yyyyMMddHH"));
		for (final String channel : channels) {
			
			threadsNumber++;
			
		}
		for (final String channel : channels) {
			
				taskExecutor.execute(new Runnable() {
					@Override
					public void run() {
						Calendar cStart = Calendar.getInstance();
						cStart.setTime(startTime);
						Calendar cEnd = Calendar.getInstance();
						cEnd.setTime(endTime);
						while (cStart.before(cEnd)) {
							
							StringBuffer startRowKey1 = new StringBuffer(RowkeyUtils.getRowKey(DateUtils.format(cStart.getTime(),
									"yyyyMMddHH"), channel,  RowkeyUtils.START_ROW_KEY));
							
							StringBuffer endRowKey1 = new StringBuffer(RowkeyUtils.getRowKey(DateUtils.format(cStart.getTime(),
									"yyyyMMddHH"), channel,  RowkeyUtils.STOP_ROW_KEY));
							
							log.info(Thread.currentThread().getName()
									+ " 历史航班价格统计 startRowKey1:"
									+ startRowKey1.toString() + " endRowKey1:"
									+ endRowKey1);
							// 统计航班 航线的每小时和每天的最低价
							processDataByHour(startRowKey1, endRowKey1);
							cStart.add(Calendar.HOUR_OF_DAY, 1);// 小时数加一
						}
	
					}
				});
			
		}

		log.info("HistoryDataPreprocess spend time(ms)：" + (System.currentTimeMillis() - start));
	}

	/*-
	 * 
	 * 充分利用hbase数据库rowkey的字典序及rowkey的设计，将相同的航班放在相邻的位置进行统计，减少内存的占用
	 * 
	 * 按小时处理每小时的最低价
	 * 	 BASE_DATA表的 rowkey设计
	 *	 抓取基础数据表(BASE_DATA)：
	 *	 抓取日期(年月日时)_渠道_区域ID_起点(使用三字码)_终点(使用三字码)_航班日期(年月日)_航班ID(8)_5位随即数
	 *	         10位 +    5位  +2位 +    3位   +        3位      + 8位+    8位 +    5位
	 * @param startRowKey
	 * @param endRowKey
	 * @throws IOException
	 * @throws Exception
	 */
	private void processDataByHour(StringBuffer startRowKey,
			StringBuffer endRowKey) {
		ResultScanner rs = null;
		try {
			rs = HBaseHelper.getRecords(HBaseTable.BASE_DATA_FLIGHT,
					startRowKey.toString(), endRowKey.toString(),new String[]{"planeinfo"},null);
			// System.out.println(" startRowKey :" + startRowKey + " endRowKey:"
			// + endRowKey);
		} catch (Exception e1) {
			try {
				Thread.sleep(60 * 1000L);
				rs = HBaseHelper.getRecords(HBaseTable.BASE_DATA_FLIGHT,
						startRowKey.toString(), endRowKey.toString(),new String[]{"planeinfo"},null);
			} catch (Exception e) {
				log.error(StringUtils.getStackTrace(e));
			}
			
		}
		String flightId = null;
		float flightLowPrice = Integer.MAX_VALUE;
		Map<String, Map<String, String>> flightFCV = new HashMap<String, Map<String, String>>();// 保存HBASE中存储的信息
		try{
			for (Result res : rs) {
				//System.out.println("currentRowkey:" + new String(res.getRow()));
				try {// 某一条航班数据的错误，不应该影响其他航班数据的预处理
					tasksNumber++;
					Map<String, Map<String, String>> currentMap = HBaseHelper
							.transformHBaseRowToMap(res);
					Map<String, String> curPlaneinfo = null;
					Map<String, String> curFlightMap = currentMap.get("planeinfo");
					float flightPrice = Float.parseFloat(curFlightMap
							.get("lowerPrice") != null ? curFlightMap
									.get("lowerPrice") : "0");
//					String channelKey=curFlightMap.get("grabChannelNum");
//					if("00006".equals(channelKey)){
//						if(PriceUtil.USDrefePrice==0){
//							PriceUtil.USDrefePrice=this.getUSDrefePrice();
//						}
//						flightPrice=PriceUtil.conversionPrice(flightPrice, PriceUtil.USDrefePrice).floatValue();
//						curFlightMap.put("lowerPrice", flightPrice+"");
//						curFlightMap.put("highPrice", PriceUtil.conversionPrice(Double.parseDouble(null==curFlightMap.get("highPrice")?"0":curFlightMap.get("highPrice")),PriceUtil.USDrefePrice)+"");
//						currentMap.put("planeinfo", curFlightMap);
//						
//					}
					if (flightPrice <= 0) {// 如果存在坏数据，就跳过
//						log.info("rowkey[" + new String(res.getRow())
//								+ "] 数据可能存在，最低价小于或等于0");
						continue;
					}
					
					String curFlightId = curFlightMap.get("flightBatchDate")
							.replace("-", "") + curFlightMap.get("flightId");
					if (flightId != null && !flightId.equalsIgnoreCase(curFlightId)) {// 如果此航班比较完毕，保存航班最低价信息
						Map<String, String> planeMap = flightFCV.get("planeinfo");
						if(planeMap.get("flightId")==null) return;
						if(planeMap.get("flightId").indexOf("|")>0){
							planeMap.put("flightId", planeMap.get("flightId").substring(0, planeMap.get("flightId").indexOf("|")));
						}
						if(planeMap.get("flightId").indexOf(",")>0){
							planeMap.put("flightId", planeMap.get("flightId").substring(0, planeMap.get("flightId").indexOf(",")));
						}
						// 与当天抓取的此航班的其他价格进行比较 重新组装 startRowKey endRowKey
						// 抓取日期(年月日时)_渠道_区域_起点(使用三字码)_终点(使用三字码)_航班日期(年月日时)_航班号_5位随即数
						Map<String, String> hisFlightMap = flightFCV
								.get("planeinfo");
						if(hisFlightMap.get("flightId")==null) return;
						if(hisFlightMap.get("flightId").indexOf("|")>0){
							hisFlightMap.put("flightId", hisFlightMap.get("flightId").substring(0, hisFlightMap.get("flightId").indexOf("|")));
						}
						if(hisFlightMap.get("flightId").indexOf(",")>0){
							hisFlightMap.put("flightId", hisFlightMap.get("flightId").substring(0, hisFlightMap.get("flightId").indexOf(",")));
						}
						
						compareHistoryFlightSameDay(planeMap, flightLowPrice);
						flightId = null;// 保存号将flight置空
						flightLowPrice = Integer.MAX_VALUE;
					}
					if (flightPrice < flightLowPrice) {// 比较航班价格 暂存航班信息
						flightLowPrice = flightPrice;
						flightFCV = currentMap;
					}
					curPlaneinfo = currentMap.get("planeinfo");
					flightId = curPlaneinfo.get("flightBatchDate").replace("-", "")
							+ curPlaneinfo.get("flightId");
				} catch (Exception e) {
					failuresNumber++;
					log.info("rowkey[" + new String(res.getRow()) + "]数据预处理错误\r\n");
				}
			}
		}catch (Exception e) {
				failuresNumber++;
				log.error(StringUtils.getStackTrace(e));
			
		}finally{
			if(rs != null) rs.close();
			rs = null;
			executeNumber++;
			if(executeNumber==threadsNumber){
				LogMessageUtils.build("统计历史最低价", this.getClass().getSimpleName(), 2, null, 
						failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime);
			}
		}
	}

	/**
	 * 比较历史的航班信息
	 * 
	 * @param startRowKey
	 * @param endRowKey
	 * @param planeMap
	 *            最新的航班价格的航班信息
	 * @param newFlightLowPrice
	 *            最新的航班价格
	 */
	private void compareHistoryFlightSameDay( Map<String, String> newPlaneMap,
			float newFlightLowPrice) {
		
		int start = Integer.parseInt(new String(newPlaneMap.get("grabDate").substring(0, 8))
				+ "00");
		int end = Integer.parseInt(new String(newPlaneMap.get("grabDate").substring(0, 10)));
		
		Map<String, String> lowestPlaneMap = newPlaneMap;// 暂存航班信息
		float flightLowPrice = newFlightLowPrice;
		for (int i = start; i < end; i++) {// 循环取出某天的航班在抓取日期的内的所有抓取到的数据的最小值
			ResultScanner rs = null;
			try {
				String newStartRowkey=RowkeyUtils.getRowKey(i+"", newPlaneMap.get("grabChannelNum"),
						 newPlaneMap.get("fromCity"), newPlaneMap.get("toCity"),
						newPlaneMap.get("flightBatchStartTime").substring(0, 10), newPlaneMap.get("flightId"), RowkeyUtils.START_ROW_KEY);
				String newStopRowkey=RowkeyUtils.getRowKey(i+"", newPlaneMap.get("grabChannelNum"),
						 newPlaneMap.get("fromCity"), newPlaneMap.get("toCity"),
						newPlaneMap.get("flightBatchStartTime").substring(0, 10), newPlaneMap.get("flightId"), RowkeyUtils.STOP_ROW_KEY);
				
				rs = HBaseHelper.getRecords(HBaseTable.BASE_DATA_FLIGHT,
						newStartRowkey,
						newStopRowkey,new String[]{"planeinfo"},null);

				for (Result res : rs) {
					try {// 某一条航班数据的错误，不应该影响其他航班数据的预处理
						Map<String, Map<String, String>> curFlightMap = HBaseHelper
								.transformHBaseRowToMap(res);
						float flightPrice = Float.parseFloat(curFlightMap.get(
								"planeinfo").get("lowerPrice"));
						
//						String channelKey=curFlightMap.get("planeinfo").get("grabChannelNum");
//						if("00006".equals(channelKey)){
//							if(PriceUtil.USDrefePrice==0){
//								PriceUtil.USDrefePrice=this.getUSDrefePrice();
//							}
//							flightPrice=PriceUtil.conversionPrice(flightPrice, PriceUtil.USDrefePrice).floatValue();
//							curFlightMap.get("planeinfo").put("lowerPrice", flightPrice+"");
//							curFlightMap.get("planeinfo").put("highPrice", PriceUtil.conversionPrice(Double.parseDouble(null==curFlightMap.get("planeinfo").get("highPrice")?"0":curFlightMap.get("planeinfo").get("highPrice")),PriceUtil.USDrefePrice)+"");
//						}
						
						if (flightPrice < flightLowPrice) {// 比较航班价格 暂存航班信息
							flightLowPrice = flightPrice;
							lowestPlaneMap = curFlightMap.get("planeinfo");
						}
					} catch (Exception e) {
						log.info("rowkey[" + new String(res.getRow())
								+ "]数据预处理错误\r\n");
					}
				}
			} catch (Exception e1) {
				log.info(StringUtils.getStackTrace(e1));
			}finally{
				if(rs != null) rs.close();
				rs = null;
			}
		}
		saveLowestFlightPrice(lowestPlaneMap);
	}

	/**
	 * 每天的航班历史最低价(即每天抓取多次航班中的最低价格)LP_FLIGHT_HISTORY rowkey:
	 * 渠道_区域ID_起点(使用三字码)_终点(使用三字码)_抓取时间(年月日)_航班日期(年月日时)_航班号
	 * 
	 * @param planeMap
	 * @throws Exception
	 */
	private void saveLowestFlightPrice(Map<String, String> planeMap) {

		// System.out.println();
		String grabChannelNum = planeMap.get("grabChannelNum");
		String fromCity = planeMap.get("fromCity");
		String toCity = planeMap.get("toCity");
		String flightId = RowkeyUtils.getFlightIdHbaseKey(planeMap
				.get("flightId"));
		String flightBatchStartTime = planeMap.get("flightBatchStartTime")
				.substring(0, 10);
		String grabDate = planeMap.get("grabDate").substring(0, 8);
		// 每天的航班历史最低价(即每天抓取多次航班中的最低价格)LP_FLIGHT_HISTORY rowkey:
		// 渠道_起点(使用三字码)_终点(使用三字码)_抓取时间(年月日)_航班日期(年月日时)_航班号
		StringBuffer sb = new StringBuffer();// 组装rowkey
		sb.append(grabChannelNum);
		//sb.append(areaCode); //新rowkey 不要区域
		sb.append(fromCity);
		sb.append(toCity);
		sb.append(grabDate);
		sb.append(flightBatchStartTime);
		sb.append(flightId);
		this.deleteTableRow(HBaseTable.TABLE_FLIGHT_HISTORY,sb.toString());
		Put put = new Put(sb.toString().getBytes());
		// System.out.println("   rowkey: " + sb.toString());
		for (Entry<String, String> en : planeMap.entrySet()) {
			put.add("planeinfo".getBytes(), en.getKey().getBytes(), en
					.getValue().getBytes());
		}
		try {
			HBaseHelper.addRecord(HBaseTable.TABLE_FLIGHT_HISTORY, put);
			insertNumber++;
		} catch (Exception e) {
			failuresNumber++;
			log.info(StringUtils.getStackTrace(e));
		}

	}
	private void deleteTableRow(String tableName, String rowkey){
		try {
			HBaseHelper.delRecord(tableName, rowkey);
		} catch (Exception e2) {
			log.error("insert before delete rowkey="+rowkey+"  "+e2.getMessage());
		}
	}
	

}
